package com.behavioranalysis.flinkprogram.dao.output;

import com.behavioranalysis.flinkprogram.conf.ConfigurationManager;
import com.behavioranalysis.flinkprogram.constant.Constants;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class AdClickTrendSink extends RichSinkFunction<Tuple5<String, String, String, Long, Long>> {
    private Connection conn = null;
    private PreparedStatement pstm = null;

    /**
     * 一、open()方法中建立连接，这样不用每次invoke的时候都要建立连接和释放连接。
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        String driver = "";
        String url = "";
        String username = "";
        String password = "";
        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            driver = ConfigurationManager.getProperty(Constants.JDBC_DRIVER);
            url = ConfigurationManager.getProperty(Constants.JDBC_URL);
            username = ConfigurationManager.getProperty(Constants.JDBC_USER);
            password = ConfigurationManager.getProperty(Constants.JDBC_PASSWORD);
        } else {
            driver = ConfigurationManager.getProperty(Constants.JDBC_DRIVER);
            url = ConfigurationManager.getProperty(Constants.JDBC_URL_PROD);
            username = ConfigurationManager.getProperty(Constants.JDBC_USER_PROD);
            password = ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD);
        }
        //1.加载驱动
        Class.forName(driver);
        //2.创建连接
        conn = DriverManager.getConnection(url, username, password);
        String sql = "insert into ad_click_trend (date,hour,minute,ad_id,click_count) values (?,?,?,?,?) on DUPLICATE key update click_count=values(click_count)";
        //3.获得执行语句
        pstm = conn.prepareStatement(sql);
    }

    /**
     * 二、每个元素的插入都要调用一次invoke()方法，这里主要进行插入操作
     *
     * @param clickCount
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(Tuple5<String, String, String, Long, Long> clickCount, Context context) throws Exception {
        try {
            //4.组装数据，执行插入操作
            pstm.setString(1, clickCount.f0);
            pstm.setString(2, clickCount.f1);
            pstm.setString(3, clickCount.f2);
            pstm.setInt(4, clickCount.f3.intValue());
            pstm.setInt(5, clickCount.f4.intValue());
            pstm.executeUpdate();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        //5.关闭连接和释放资源
        if (conn != null) {
            conn.close();
        }
        if (pstm != null) {
            pstm.close();
        }
    }
}
